package cn.net.yugu.doraemon.datahub.core;

import cn.net.yugu.doraemon.datahub.config.DataHubProperties;
import cn.net.yugu.doraemon.datahub.worker.SendRecordWorker;
import cn.net.yugu.doraemon.datahub.model.SendResult;
import com.aliyun.datahub.client.DatahubClient;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wanghongli
 * @date 2024/5/29 15:44
 * @description DataHubProducer
 **/
@Slf4j
public class DataHubTemplate implements InitializingBean, DisposableBean {
    private ConcurrentHashMap<String, SendRecordWorker> recordSenderMap;
    private final DatahubClient datahubClient;
    private final DataHubProperties dataHubProperties;

    public DataHubTemplate(DatahubClient datahubClient, DataHubProperties dataHubProperties) {
        this.datahubClient = datahubClient;
        this.dataHubProperties = dataHubProperties;
    }

    @Override
    public void destroy() throws Exception {
        if (recordSenderMap.isEmpty()) {
            return;
        }
        recordSenderMap.forEach((k, v) -> v.stop());
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        recordSenderMap = new ConcurrentHashMap<>(256);
    }

    public SendResult send(String topicName, Object data) {
        return batchSend(topicName, Lists.newArrayList(data));
    }

    public SendResult batchSend(String topicName, List<Object> dataList) {
        final String projectName = dataHubProperties.getProjectName();
        if (CollectionUtils.isEmpty(dataList) || StringUtils.isBlank(projectName) || StringUtils.isBlank(topicName)) {
            return SendResult.build(0,0, "参数不正确");
        }
        return SendRecordWorker.sendDataHubRecords(this.datahubClient, projectName, topicName, dataList);
    }

    public void asyncSend(String topicName, Object data) {
        asyncBatchSend(topicName, Lists.newArrayList(data));
    }

    public void asyncBatchSend(String topicName, List<Object> dataList) {
        final String projectName = dataHubProperties.getProjectName();
        if (CollectionUtils.isEmpty(dataList) || StringUtils.isBlank(projectName) || StringUtils.isBlank(topicName)) {
            return;
        }
        final String recordNm = (projectName + ":" + topicName).intern();
        SendRecordWorker sendRecordWorker = recordSenderMap.get(recordNm);
        if (sendRecordWorker == null) {
            synchronized (this) {
                sendRecordWorker = recordSenderMap.get(recordNm);
                if (sendRecordWorker == null) {
                    sendRecordWorker = new SendRecordWorker(datahubClient, projectName, topicName, dataHubProperties.getMinBatchSendSize(), dataHubProperties.getMaxBatchSendSize(), dataHubProperties.getMaxSendSizePerRequest(), dataHubProperties.getMaxBatchSendLingerMs());
                    recordSenderMap.put(recordNm, sendRecordWorker);
                }
            }
        }
        sendRecordWorker.addRecords(dataList);
    }
}
